package com.cl.spark.nodenew;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Component
public class UnionTableNode extends BaseSparkNode {
    @Override

    public SparkResult process(SparkParam sparkParam) {
        JSONArray res = new JSONArray();
        Set<String> keys = new HashSet<>();
        List<JSONArray> dataList = sparkParam.getDataList();
        for (int i = 0; i < dataList.size(); i++) {
            if (dataList.get(i).isEmpty()) {
                continue;
            }
            keys.addAll(dataList.get(i).getJSONObject(0).keySet());
            res.addAll(dataList.get(i));
        }
        for (int i = 0; i < res.size(); i++) {
            JSONObject item = res.getJSONObject(i);
            keys.forEach(key -> item.putIfAbsent(key, null));
            res.set(i, item);
        }
        return SparkResult.success(res);
    }

    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.UNION_TABLE;
    }
}
